package reports

import (
	"bytes"
	"context"
	"database/sql"
	"errors"
	"fmt"
	"strconv"
	"strings"
	"time"

	"shqsoft.cn/pas3ui/base"
)

var ctx = context.TODO()

type DataBase struct {
	tableName    string
	config       *Config
	db           *sql.DB
	insSQL       string
	mustColNames []string
	mustColExps  []string
	leafCols     []*Column
}

func NewDataBase(tName string, conf *Config) (db *DataBase, err error) {
	db = &DataBase{
		tableName: tName,
		config:    conf,
	}
	if conf.DataSource.DbName == "ch" || conf.DataSource.DbName == "" {
		db.db, err = base.CHDB.DB()
	} else if conf.DataSource.DbName == "my" {
		db.db, err = base.MYDB.DB()
	} else {
		err = fmt.Errorf("数据源中的数据库名称%s错误，必须是ch或者my", conf.DataSource.DbName)
	}
	if err != nil {
		return
	}
	mustColNames := make([]string, 0, 10)
	mustColExps := make([]string, 0, 10)
	var buff bytes.Buffer
	buff.WriteString("insert into ")
	buff.WriteString(tName)
	buff.WriteString("(")
	for _, col := range conf.Columns {
		buff.WriteString(col.Name)
		buff.WriteString(",")
		if col.DataType == "float" || len(col.Children) > 0 {
			continue
		}
		colExp := col.ColExp
		if colExp == "" {
			colExp = col.Name
		}
		mustColNames = append(mustColNames, col.Name)
		mustColExps = append(mustColExps, fmt.Sprintf("%s AS %s", colExp, col.Name))
	}
	buff.Truncate(buff.Len() - 1)
	buff.WriteString(") values(")
	for i := 0; i < len(conf.Columns); i++ {
		buff.WriteString("?,")
	}
	buff.Truncate(buff.Len() - 1)
	buff.WriteString(")")
	db.insSQL = buff.String()
	db.mustColNames = mustColNames
	db.mustColExps = mustColExps
	db.leafCols, err = conf.GetLeafColumns()
	return
}
func (d *DataBase) tableTxists() (exists bool, err error) {
	var sqlStr string
	if d.config.DataSource.DbName == "ch" {
		sqlStr = fmt.Sprintf("select count(*) from system.tables c where database ='%s' and name = ?", base.AppConf.CHDBName)
	} else {
		sqlStr = "select count(*) from information_schema.tables where TABLE_NAME = ? "
	}
	var count int
	err = d.db.QueryRow(sqlStr, d.tableName).Scan(&count)
	if err != nil {
		return
	}
	if count > 0 {
		exists = true
	}
	return
}
func (d *DataBase) createTable(engine string) (exists bool, err error) {
	if engine != "Memory" && engine != "SummingMergeTree" {
		err = fmt.Errorf("表引擎%s无效，只支持Memory和SummingMergeTree两种", engine)
		return
	}
	isSum := false
	if engine == "SummingMergeTree" {
		isSum = true
	}
	exists, err = d.tableTxists()
	if err != nil {
		return
	}
	if exists {
		return
	}
	var buff bytes.Buffer
	buff.WriteString("CREATE TABLE ")
	buff.WriteString(d.tableName)
	buff.WriteString("(")
	calcColNames := make([]string, 0, 20)
	for _, col := range d.leafCols {
		if col.Name == "" {
			err = fmt.Errorf("报表%s的column定义有误，没有定义name", d.config.Name)
			return
		}
		dType := getColDBType(d.config.DataSource.DbName, col.DataType, col.Width)
		buff.WriteString(col.Name)
		buff.WriteString(" ")
		buff.WriteString(dType)
		buff.WriteString(",")
		for _, cc := range d.config.CalcCols {
			if base.InStringSlice(col.GroupCalc, cc.RefColTypes) && !base.InStringSlice(col.Name, cc.ExcludeCols) {
				colName := fmt.Sprintf("%s_%s", col.Name, cc.Name[1:])
				buff.WriteString(colName)
				buff.WriteString(" ")
				buff.WriteString(dType)
				buff.WriteString(",")
				calcColNames = append(calcColNames, colName)
			}
		}
	}
	buff.WriteString("gp_index ")
	if d.config.DataSource.DbName == "ch" {
		buff.WriteString("UInt8")
	} else {
		buff.WriteString("tinyint(1)")
	}
	buff.WriteString(")")
	colLevBy := make([]string, 0, 4)
	for _, clev := range d.config.ColumnLevels {
		colLevBy = append(colLevBy, clev.IdCol)
	}
	if d.config.DataSource.DbName == "ch" {
		buff.WriteString(" ENGINE = ")
		buff.WriteString(engine)
		if engine == "SummingMergeTree" {
			buff.WriteString("((")
			has := false
			for _, col := range d.leafCols {
				if col.GroupCalc == "acbl" || col.GroupCalc == "avg" || col.GroupCalc == "sum" || (strings.Contains(col.ColExp, "sum") && col.GroupExp == "") {
					has = true
					buff.WriteString(col.Name)
					buff.WriteString(",")
				}
			}
			for _, ccName := range calcColNames {
				buff.WriteString(ccName)
				buff.WriteString(",")
			}
			if has {
				buff.Truncate(buff.Len() - 1)
			}
			buff.WriteString("))")
			buff.WriteString(" ORDER BY (")
			buff.WriteString(strings.Join(d.config.KeyColumns, ","))
			if len(colLevBy) > 0 {
				buff.WriteString(",")
				buff.WriteString(strings.Join(colLevBy, ","))
			}
			buff.WriteString(")")
		} else {
			buff.WriteString("()")
		}
	} else {
		if isSum {
			err = errors.New("mysql不支持SummingMergeTree引擎")
			return
		}
		buff.WriteString(" ENGINE = ")
		buff.WriteString(engine)
	}
	_, err = d.db.Exec(buff.String())
	return
}
func (d *DataBase) createColumn(col *Column, calcName string, sdate, edate *time.Time, cynoWhere string, paramVals map[string]interface{}) (err error) {
	colName := fmt.Sprintf("%s_%s", col.Name, calcName[1:])
	skey := fmt.Sprintf("smt:%s", d.tableName)
	isHas := base.RedisCli.SIsMember(ctx, skey, colName).Val()
	if isHas {
		return
	}
	gp_index := len(d.config.Groups)
	var groupBy string
	if gp_index == 0 {
		groupBy = strings.Join(d.config.KeyColumns, ",")
	} else {
		groupBy = strings.Join(d.config.Groups[gp_index-1].By, ",")
	}
	mustColNameStr := strings.Join(d.mustColNames, ",")
	strSQl := fmt.Sprintf("insert into %s (%s,%s,gp_index) select %s,%s,gp_index from %s where gp_index=?", d.tableName, mustColNameStr, colName, mustColNameStr, col.Name, d.tableName)
	_, err = d.db.Exec(strSQl, gp_index)
	if err != nil {
		return
	}
	colExp := col.ColExp
	if colExp == "" {
		colExp = col.Name
	} else if strings.HasPrefix(colExp, "@") {
		pName := colExp[1:]
		if pv, ok := paramVals[pName]; ok {
			colExp = base.ToString(pv)
		} else {
			err = fmt.Errorf("字段%s指定的参数colExp:%s无效", col.Name, colExp)
			return
		}
	}
	var date time.Time
	if edate == nil {
		date = *sdate
	} else {
		date = *edate
	}
	if col.GroupCalc == "acbl" {
		bsnjs := false
		switch calcName {
		case "#bsr":
			date = date.AddDate(0, 0, -1)
		case "#bsy":
			date = time.Date(date.Year(), date.Month(), 1, 0, 0, 0, 0, time.Local)
			date = date.AddDate(0, 0, -1)
		case "#bsj":
			month := date.Month()
			switch month {
			case 2, 5, 7, 11:
				date = date.AddDate(0, -1, 0)
			case 3, 6, 9, 12:
				date = date.AddDate(0, -2, 0)
			}
			date = time.Date(date.Year(), date.Month(), 1, 0, 0, 0, 0, time.Local)
			date = date.AddDate(0, 0, -1)
		case "#bsn":
			date = time.Date(date.Year(), 1, 1, 0, 0, 0, 0, time.Local)
			date = date.AddDate(0, 0, -1)
		case "#bsntq":
			date = time.Date(date.Year()-1, date.Month(), date.Day(), 0, 0, 0, 0, time.Local)
		case "#bsnjs":
			bsnjs = true
		default:
			err = fmt.Errorf("生成列%s上的比上日计算出错，不支持的计算%s", col.Name, calcName)
			return
		}
		if bsnjs {
			err = d.createBaseColumn(date, colName, "ACBL", cynoWhere, groupBy, gp_index)
		} else {
			err = d.createAcblColumn(date, colName, colExp, cynoWhere, groupBy, gp_index)
		}
		if err != nil {
			return err
		}
	} else if col.GroupCalc == "avg" || col.GroupCalc == "sum" {
		var st, et time.Time
		days := 1
		switch calcName {
		case "#bsr":
			err = fmt.Errorf("生成列%s上的比上日计算出错，日均和合计算法不适用于比上日", col.Name)
			return
		case "#bsy":
			et = time.Date(date.Year(), date.Month(), 1, 0, 0, 0, 0, time.Local)
			et = et.AddDate(0, 0, -1)
			st = time.Date(et.Year(), et.Month(), 1, 0, 0, 0, 0, time.Local)
			if col.GroupCalc == "avg" {
				days = int(et.Sub(st).Hours())/24 + 1
			}
			smont, _ := strconv.Atoi(st.Format("200601"))
			emont, _ := strconv.Atoi(et.Format("200601"))
			err = d.createSumColumnFormMF(smont, emont, colName, colExp, cynoWhere, groupBy, gp_index, days)
			if err != nil {
				return err
			}
		case "#bsj":
			month := date.Month()
			switch month {
			case 2, 5, 7, 11:
				date = date.AddDate(0, -1, 0)
			case 3, 6, 9, 12:
				date = date.AddDate(0, -2, 0)
			}
			et = time.Date(date.Year(), date.Month(), 1, 0, 0, 0, 0, time.Local)
			st = et.AddDate(0, -3, 0)
			et = et.AddDate(0, 0, -1)
			if col.GroupCalc == "avg" {
				days = int(et.Sub(st).Hours())/24 + 1
			}
			smont, _ := strconv.Atoi(st.Format("200601"))
			emont, _ := strconv.Atoi(et.Format("200601"))
			err = d.createSumColumnFormMF(smont, emont, colName, colExp, cynoWhere, groupBy, gp_index, days)
			if err != nil {
				return err
			}
		case "#bsn":
			et = time.Date(date.Year(), 1, 1, 0, 0, 0, 0, time.Local)
			st = et.AddDate(-1, 0, 0)
			et = et.AddDate(0, 0, -1)
			if col.GroupCalc == "avg" {
				days = int(et.Sub(st).Hours())/24 + 1
			}
			smont, _ := strconv.Atoi(st.Format("200601"))
			emont, _ := strconv.Atoi(et.Format("200601"))
			err = d.createSumColumnFormMF(smont, emont, colName, colExp, cynoWhere, groupBy, gp_index, days)
			if err != nil {
				return err
			}
		case "#bsntq":
			et = date.AddDate(-1, 0, 0)
			if date.Unix() == sdate.Unix() {
				st = time.Date(et.Year(), 1, 1, 0, 0, 0, 0, time.Local)
			} else {
				st = sdate.AddDate(-1, 0, 0)
			}
			if col.GroupCalc == "avg" {
				days = int(et.Sub(st).Hours())/24 + 1
			}
			dr, mr, err := parsesDateRange(&st, &et)
			if err != nil {
				return err
			}
			for _, r := range dr {
				err = d.createSumColumnFormDF(r.sdate, r.edate, colName, colExp, cynoWhere, groupBy, gp_index, days)
				if err != nil {
					return err
				}
			}
			if mr != nil {
				smont, _ := strconv.Atoi(mr.smon)
				emont, _ := strconv.Atoi(mr.emon)
				err = d.createSumColumnFormMF(smont, emont, colName, colExp, cynoWhere, groupBy, gp_index, days)
				if err != nil {
					return err
				}
			}
		case "#bsnjs":
			err = d.createBaseColumn(date, colName, "AVVL", cynoWhere, groupBy, gp_index)
			if err != nil {
				return
			}
		default:
			err = fmt.Errorf("生成列%s上的比上日计算出错，不支持的计算%s", col.Name, calcName)
			return
		}
	}
	err = d.createColumnGroupBy(colName)
	if err != nil {
		return
	}
	base.RedisCli.SAdd(ctx, skey, colName)
	base.RedisCli.Expire(ctx, skey, time.Hour*8)
	return
}
func (d *DataBase) createAcblColumn(date time.Time, colName, colExp, cynoWhere, groupby string, gp_index int) (err error) {
	if cynoWhere != "" {
		cynoWhere = " AND " + cynoWhere
	}
	dtid := date.Format("2006-01-02")
	strSQl := fmt.Sprintf("insert into %s (%s,%s,gp_index) select %s,-sum(%s),? from %s where DTID=? %s group by %s", d.tableName, strings.Join(d.mustColNames, ","), colName, strings.Join(d.mustColExps, ","), colExp, d.config.DataSource.DFTable, cynoWhere, groupby)
	_, err = d.db.Exec(strSQl, gp_index, dtid)
	return
}
func (d *DataBase) createBaseColumn(date time.Time, colName, bType, cynoWhere, groupby string, gp_index int) (err error) {
	if bType != "ACBL" && bType != "AVVL" {
		return fmt.Errorf("参数bType:%s必须是ACBL或者AVVL", bType)
	}
	dtid := date.Format("2006-01-02")
	dateStr := date.Format("20060102")
	cname := strings.TrimSuffix(colName, "_bsnjs")
	bcol := base.GetBigTabCol(cname)
	if bcol == nil {
		return fmt.Errorf("列%s在bigTable配置中不存在", colName)
	}
	var tName string
	switch bcol.DiviTable {
	case "T_COMP_ACCO_DIVI_ZIPPER":
		tName = fmt.Sprintf("TMP_COMP_BASE_%s", dateStr)
		err = createCompBase(&date)
	case "T_INDI_ACCO_DIVI_ZIPPER":
		tName = fmt.Sprintf("TMP_INDI_SAVE_BASE_%s", dateStr)
		err = createIndiSaveBase(&date)
	case "T_INDI_LOAN_DIVI_ZIPPER":
		tName = fmt.Sprintf("TMP_INDI_LOAN_BASE_%s", dateStr)
		err = createIndiLoanBase(&date)
	case "TMP_INDI_CLIE_DIVI":
		tName = fmt.Sprintf("TMP_INDI_CLIE_BASE_%s", dateStr)
		err = createIndiClieBase(&date)
	default:
		return fmt.Errorf("列%s在bigTable配置中的DiviTable:%s无效", colName, bcol.DiviTable)
	}
	if err != nil {
		return
	}
	colExp := bcol.BaseColName
	if colExp == "" {
		return fmt.Errorf("列%s在bigTable中没有配置baseColName", colName)
	}
	if bType == "AVVL" && colExp == "ACBL" {
		colExp = "AVVL"
	}
	whereStr := bcol.BaseWhere
	if whereStr != "" {
		whereStr = " WHERE " + whereStr
		if bcol.Where != "" {
			whereStr = whereStr + " AND " + bcol.Where
		}
		if cynoWhere != "" {
			whereStr = whereStr + " AND " + cynoWhere
		}
	} else {
		if bcol.Where != "" {
			whereStr = " WHERE " + bcol.Where
			if cynoWhere != "" {
				whereStr = whereStr + " AND " + cynoWhere
			}
		} else {
			if cynoWhere != "" {
				whereStr = " WHERE " + cynoWhere
			}
		}
	}
	strSQl := fmt.Sprintf(`insert into %s (%s,%s,gp_index)
	select %s,-sum(%s),? from %s %s group by %s`, d.tableName, strings.Join(d.mustColNames, ","), colName, strings.Join(d.mustColExps, ","), colExp, tName, whereStr, groupby)
	_, err = d.db.Exec(strSQl, gp_index, dtid)
	return
}
func (d *DataBase) createSumColumnFormDF(sdate, edate string, colName, colExp, cynoWhere, groupby string, gp_index int, days int) (err error) {
	if cynoWhere != "" {
		cynoWhere = " AND " + cynoWhere
	}
	strSQl := fmt.Sprintf("insert into %s (%s,%s,gp_index) select %s,-sum(%s)/?,? from %s where DTID>=? and DTID<=? %s group by %s",
		d.tableName, strings.Join(d.mustColNames, ","), colName, strings.Join(d.mustColExps, ","), colExp, d.config.DataSource.DFTable,
		cynoWhere, groupby)
	_, err = d.db.Exec(strSQl, days, gp_index, sdate, edate)
	return
}
func (d *DataBase) createSumColumnFormMF(smont, emont int, colName, colExp, cynoWhere, groupby string, gp_index int, days int) (err error) {
	if cynoWhere != "" {
		cynoWhere = " AND " + cynoWhere
	}
	strSQl := fmt.Sprintf("insert into %s (%s,%s,gp_index) select %s,-sumMerge(%s)/?,? from %s where MONT>=? and MONT<=? %s group by %s",
		d.tableName, strings.Join(d.mustColNames, ","), colName, strings.Join(d.mustColExps, ","), colExp,
		d.config.DataSource.MFTable, cynoWhere, groupby)
	_, err = d.db.Exec(strSQl, days, gp_index, smont, emont)
	return
}
func (d *DataBase) createColumnGroupBy(colName string) (err error) {
	if len(d.config.Groups) == 0 {
		return
	}
	var listIdx int
	for i, g := range d.config.Groups {
		if len(g.ExcludeCols) == 0 {
			listIdx = i + 1
			break
		}
	}
	for i, g := range d.config.Groups {
		if len(g.ExcludeCols) == 0 {
			continue
		}
		groupBy := strings.Join(g.By, ",")
		var buffIns bytes.Buffer
		buffIns.WriteString("insert into ")
		buffIns.WriteString(d.tableName)
		buffIns.WriteString("(")
		buffIns.WriteString(strings.Join(d.mustColNames, ","))
		buffIns.WriteString(",")
		buffIns.WriteString(colName)
		buffIns.WriteString(",gp_index)")
		buffIns.WriteString(" select ")
		for _, mcn := range d.mustColNames {
			if base.InStringSlice(mcn, g.ExcludeCols) {
				var col *Column
				for _, c := range d.config.Columns {
					if c.Name == mcn {
						col = &c
						break
					}
				}
				if col == nil {
					err = fmt.Errorf("报表%s分组%s的ExcludeCols中指定的字段名%s在Columns中没有找到", d.config.Name, g.Title, mcn)
					return
				}
				if col.DataType == "date" {
					buffIns.WriteString("'1970-01-01'")
				} else if col.DataType == "int" || col.DataType == "float" {
					buffIns.WriteString("-1")
				} else {
					buffIns.WriteString("''")
				}
			} else {
				if base.InStringSlice(mcn, g.By) {
					buffIns.WriteString(mcn)
				} else {
					buffIns.WriteString("anyLast(")
					buffIns.WriteString(mcn)
					buffIns.WriteString(")")
				}
			}
			buffIns.WriteString(",")
		}
		buffIns.WriteString("sum(")
		buffIns.WriteString(colName)
		buffIns.WriteString("),? from ")
		buffIns.WriteString(d.tableName)
		buffIns.WriteString(" where gp_index=?")
		buffIns.WriteString(" group by ")
		buffIns.WriteString(groupBy)
		_, err = d.db.Exec(buffIns.String(), i+1, listIdx)
		if err != nil {
			return
		}
	}
	return
}
func (d *DataBase) createGroupBy(days int) (err error) {
	if len(d.config.Groups) == 0 {
		return
	}
	var buffIns bytes.Buffer
	buffIns.WriteString("insert into ")
	buffIns.WriteString(d.tableName)
	buffIns.WriteString("(")
	for _, col := range d.leafCols {
		if col.Name == "" {
			err = fmt.Errorf("报表%s的column定义有误，没有定义name", d.config.Name)
			return
		}
		buffIns.WriteString(col.Name)
		buffIns.WriteString(",")
	}
	buffIns.WriteString("gp_index)")
	var listIdx int
	for i, g := range d.config.Groups {
		if len(g.ExcludeCols) == 0 {
			listIdx = i + 1
			break
		}
	}
	colLevBy := make([]string, 0, 6)
	for _, clev := range d.config.ColumnLevels {
		colLevBy = append(colLevBy, clev.IdCol)
	}
	for i, g := range d.config.Groups {
		if i == len(d.config.Groups)-1 {
			continue
		}
		var buffsel bytes.Buffer
		buffsel.WriteString(" select ")
		for _, col := range d.leafCols {
			isex := false
			for _, c := range g.ExcludeCols {
				if col.Name == c {
					isex = true
					break
				}
			}
			isBy := false
			for _, c := range g.By {
				if col.Name == c {
					isBy = true
					break
				}
			}
			if !isBy && base.InStringSlice(col.Name, colLevBy) {
				isBy = true
			}
			if isex {
				if col.DataType == "date" {
					buffsel.WriteString("'1970-01-01'")
				} else if col.DataType == "int" || col.DataType == "float" {
					buffsel.WriteString("-1")
				} else {
					buffsel.WriteString("''")
				}
				buffsel.WriteString(",")
			} else if isBy {
				buffsel.WriteString(col.Name)
				buffsel.WriteString(",")
			} else if col.GroupExp != "" {
				buffsel.WriteString(strings.ReplaceAll(col.GroupExp, "#days", strconv.Itoa(days)))
				buffsel.WriteString(",")
			} else {
				if col.DataType == "float" {
					buffsel.WriteString("sum(")
				} else {
					buffsel.WriteString("min(")
				}
				buffsel.WriteString(col.Name)
				buffsel.WriteString("),")
			}
		}
		buffsel.WriteString("? from ")
		buffsel.WriteString(d.tableName)
		buffsel.WriteString(" where gp_index=?")
		buffsel.WriteString(" group by ")
		buffsel.WriteString(strings.Join(g.By, ","))
		if len(colLevBy) > 0 {
			buffsel.WriteString(",")
			buffsel.WriteString(strings.Join(colLevBy, ","))
		}
		var buffSql bytes.Buffer
		buffSql.WriteString(buffIns.String())
		buffSql.WriteString(buffsel.String())
		_, err = d.db.Exec(buffSql.String(), i+1, listIdx)
		if err != nil {
			return
		}
	}
	return
}
func getColDBType(dbType, dataType string, width int) (colType string) {
	if dbType == "ch" {
		switch dataType {
		case "int":
			colType = "Int64"
		case "float":
			colType = "Float64"
		case "date":
			colType = "Date"
		default:
			colType = "String"
		}
	} else {
		switch dataType {
		case "int":
			colType = "int(8)"
		case "float":
			colType = "FLOAT"
		case "date":
			colType = "DATE"
		default:
			if width == 0 {
				width = 100
			}
			colType = fmt.Sprintf("VARCHAR(%d)", width/4)
		}
	}
	return
}
func (d *DataBase) insertData(tName, where string, whereParamVals []interface{}, order string) (err error) {
	var buff bytes.Buffer
	buff.WriteString("insert into ")
	buff.WriteString(d.tableName)
	buff.WriteString("(")
	for _, col := range d.leafCols {
		buff.WriteString(col.Name)
		buff.WriteString(",")
	}
	buff.WriteString("gp_index")
	buff.WriteString(")")
	buff.WriteString(" select ")
	for _, col := range d.leafCols {
		buff.WriteString(col.Name)
		buff.WriteString(",")
	}
	groupBy := len(d.config.Groups)
	buff.WriteString(strconv.Itoa(groupBy))
	buff.WriteString(" from ")
	buff.WriteString(tName)
	if where != "" {
		buff.WriteString(" where ")
		buff.WriteString(where)
	}
	if order != "" {
		buff.WriteString(" order by  ")
		buff.WriteString(order)
	}
	if len(whereParamVals) > 0 {
		_, err = d.db.Exec(buff.String(), whereParamVals)
	} else {
		_, err = d.db.Exec(buff.String())
	}
	return
}
func (d *DataBase) writeData(rows *sql.Rows) (err error) {
	if rows == nil {
		return
	}
	var colNames []string
	colNames, err = rows.Columns()
	if err != nil {
		return
	}
	idxMap := make(map[int]int, len(d.config.Columns))
	var find bool
	for j, col := range d.config.Columns {
		find = false
		for i, cName := range colNames {
			if strings.EqualFold(cName, col.Name) {
				idxMap[j] = i
				find = true
				break
			}
		}
		if !find {
			err = fmt.Errorf("配置中指定的字段名%s在数据源中不存在", col.Name)
			return
		}
	}
	tx, err := d.db.Begin()
	if err != nil {
		return err
	}
	stmt, err := tx.Prepare(d.insSQL)
	if err != nil {
		return err
	}
	clen := len(colNames)
	i := 0
	var vals []interface{}
	for rows.Next() {
		values := make([]sql.NullString, clen)
		scanArgs := make([]interface{}, clen)
		for i := 0; i < clen; i++ {
			scanArgs[i] = &values[i]
		}
		err = rows.Scan(scanArgs...)
		if err != nil {
			tx.Rollback()
			return
		}
		vals, err = d.convertValues(values, idxMap)
		if err != nil {
			tx.Rollback()
			return
		}
		_, err = stmt.Exec(vals...)
		if err != nil {
			tx.Rollback()
			return
		}
		i++
		if i%base.AppConf.CHBatchCount == 0 {
			err = tx.Commit()
			if err != nil {
				return err
			}
			stmt.Close()
			tx, err = d.db.Begin()
			if err != nil {
				return err
			}
			stmt, err = tx.Prepare(d.insSQL)
			if err != nil {
				tx.Rollback()
				return err
			}
		}
	}
	if i > 0 {
		err = tx.Commit()
		if err != nil {
			return err
		}
		stmt.Close()
	}
	return
}
func (d *DataBase) convertValues(dbVals []sql.NullString, idxMap map[int]int) (vals []interface{}, err error) {
	vals = make([]interface{}, len(d.config.Columns))
	var valStr string
	for i, col := range d.config.Columns {
		val := dbVals[idxMap[i]]
		switch col.DataType {
		case "int":
			if val.Valid {
				if val.String == "true" {
					vals[i] = 1
				} else if val.String == "false" {
					vals[i] = 0
				} else {
					ind := strings.Index(val.String, ".")
					if ind > 0 {
						dec, err := strconv.Atoi(val.String[ind+1:])
						if err != nil {
							err = fmt.Errorf("表%s字段%s,值%s不是有效的整数", d.tableName, col.Name, val.String)
							return nil, err
						}
						if dec != 0 {
							err = fmt.Errorf("表%s字段%s,值%s不是有效的整数", d.tableName, col.Name, val.String)
							return nil, err
						}
						val.String = val.String[0:ind]
					}
					valStr = strings.TrimSpace(val.String)
					if valStr == "" {
						valStr = "0"
					}
					vals[i], err = strconv.Atoi(valStr)
					if err != nil {
						err = fmt.Errorf("表%s字段%s,值%s不是有效的整数", d.tableName, col.Name, valStr)
						return
					}
				}
			} else {
				vals[i] = 0
			}
		case "float":
			if val.Valid {
				vals[i], err = strconv.ParseFloat(val.String, 64)
				if err != nil {
					err = fmt.Errorf("表%s字段%s,值%s不是有效的数值", d.tableName, col.Name, val.String)
					return
				}
			} else {
				vals[i] = 0.0
			}
		case "date":
			if val.Valid && val.String != "0" {
				if len(val.String) == 8 {
					vals[i], err = time.Parse("20060102", val.String)
				} else if len(val.String) == 10 {
					vals[i], err = time.Parse("2006-01-02", val.String)
				} else {
					if strings.HasPrefix(val.String, "1900") {
						vals[i], err = time.Parse("2006-01-02", "1970-01-01")
					} else {
						vals[i], err = time.Parse(time.RFC3339, val.String)
					}
				}
			} else {
				vals[i], err = time.Parse("2006-01-02", "1970-01-01")
			}
			if err != nil {
				err = fmt.Errorf("表%s字段%s,值%s不是有效的日期", d.tableName, col.Name, val.String)
				return
			}
		default:
			if val.Valid {
				vals[i] = val.String
			} else {
				vals[i] = ""
			}
		}
	}
	return
}

func TableExists(tableName string) (exists bool, err error) {
	sqlStr := fmt.Sprintf("select count(*) from system.tables c where database ='%s' and name = ?", base.AppConf.CHDBName)
	var count int
	err = base.CHDB.Raw(sqlStr, tableName).Scan(&count).Error
	if err != nil {
		return
	}
	if count > 0 {
		exists = true
	}
	return
}
